package com.aaa.rabbitmq;
// 消费者代码
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;

public class SimpleConsumer {
    // 使用与生产者相同的队列名
    private final static String QUEUE_NAME = "hello_queue";

    public static void main(String[] argv) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置RabbitMQ服务器的地址
        factory.setHost("localhost");
        factory.setPort(5672); // 显式指定AMQP端口

        // 创建连接和通道
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明队列（确保队列存在）
        // 注意：如果生产者已经创建了队列，这里可以省略，但为了安全起见，通常会重复声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        System.out.println(" [消费者] 等待接收消息...");

        // 创建消息接收回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            // 将消息体转换为字符串
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [消费者] 接收到消息: '" + message + "'");

            // 模拟处理消息的工作
            try {
                doWork(message);
            } finally {
                System.out.println(" [消费者] 消息处理完成");

                // 手动确认消息已被处理
                // 参数说明:
                // 1. deliveryTag: 消息的标识
                // 2. multiple: 是否批量确认
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        // 设置消费者参数并开始消费
        // 参数说明:
        // 1. queue: 队列名称
        // 2. autoAck: 是否自动确认，false表示需要手动确认
        // 3. deliverCallback: 消息投递的回调
        // 4. cancelCallback: 消费者取消的回调
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }

    // 模拟处理消息的方法
    private static void doWork(String message) {
        try {
            // 模拟处理消息需要的时间
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}